package com.cloud.message;

import com.cloud.exception.CustomException;
import com.cloud.utils.JSONUtils;
import org.apache.commons.lang3.ObjectUtils;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;

/**
 * Mqtt消息处理器
 *
 * @author yzj
 */
public abstract class AbstractQTMessage<T> extends AbstractQTConsumer {

    private static Logger logger = LoggerFactory.getLogger(AbstractQTMessage.class);


    public AbstractQTMessage() {
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        logger.debug("MQTT接收主题:{},级别:{}", topic, message.getQos());
        T data = parseMessage(message);
        if (ObjectUtils.isNotEmpty(data) && !process(data, message)) {
            logger.warn("MQTT消费失败,请求重新消费主题:{},级别:{}", topic, message.getQos());
        }
    }


    /**
     * 继承这个方法处理消息
     * 注意:消息处理方法内用try监听异常,否则Mqtt检测到异常会直接断开客户端连接
     *
     * @param object  消息范型
     * @param message 存放消息附加属性的MqttMessage
     * @return 处理结果
     */
    public abstract boolean process(T object, MqttMessage message);

    /**
     * 反序列化解析消息
     *
     * @param message 消息体
     * @return 序列化结果
     */
    protected T parseMessage(MqttMessage message) {
        if (ObjectUtils.isEmpty(message) || ObjectUtils.isEmpty(message.getPayload())) {
            return null;
        }
        final Type type = this.getMessageType();
        if (type instanceof Class) {
            try {
                return JSONUtils.toObject(new String(message.getPayload()), (Class<T>) type);
            } catch (Exception e) {
                logger.error("解析消息对象失败:{}", e.getMessage());
            }
        } else {
            logger.warn("解析消息错误.{}", message);
        }
        return null;
    }

    protected Type getMessageType() {
        Type superType = this.getClass().getGenericSuperclass();
        if (superType instanceof ParameterizedType) {
            ParameterizedType parameType = (ParameterizedType) superType;
            Type[] actualTypes = parameType.getActualTypeArguments();
            if (actualTypes.length == 1) {
                return actualTypes[0];
            } else {
                throw new CustomException("类型参数的数量必须为1");
            }
        } else {
            // 如果没有定义泛型，解析为Object
            return Object.class;
        }
    }

}
